package cn.ultrarisk.scadmin.manager.scheduler;

import cn.ultrarisk.mybatis.extension.query.Query;
import cn.ultrarisk.scadmin.manager.biz.common.context.OperationType;
import cn.ultrarisk.scadmin.manager.biz.common.service.OperationLogService;
import cn.ultrarisk.scadmin.manager.biz.monitor.service.SolrCacheMonitorService;
import cn.ultrarisk.scadmin.manager.biz.monitor.service.SolrNodeMonitorService;
import cn.ultrarisk.scadmin.manager.biz.monitor.service.SolrReplicaMonitorService;
import cn.ultrarisk.scadmin.manager.biz.solr.context.SolrState;
import cn.ultrarisk.scadmin.manager.biz.solr.dao.SolrCollectionDAO;
import cn.ultrarisk.scadmin.manager.biz.solr.dao.SolrNodeDAO;
import cn.ultrarisk.scadmin.manager.biz.solr.dao.SolrReplicaDAO;
import cn.ultrarisk.scadmin.manager.biz.solr.dao.SolrShardDAO;
import cn.ultrarisk.scadmin.manager.biz.solr.entity.*;
import cn.ultrarisk.scadmin.manager.biz.solr.service.*;
import cn.ultrarisk.scadmin.manager.biz.zookeeper.dao.ZkNodeDAO;
import cn.ultrarisk.scadmin.manager.biz.zookeeper.entity.ZkNode;
import cn.ultrarisk.scadmin.manager.biz.zookeeper.service.ZkNodeService;
import cn.ultrarisk.scadmin.manager.util.CollectionUtil;
import cn.ultrarisk.scadmin.manager.util.IOUtil;
import cn.ultrarisk.scadmin.manager.util.JsonUtil;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.net.URL;
import java.net.URLConnection;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * GatherScheduler
 *
 * @author sevendlong(lin23871@163.com)
 * @date 2014/5/17
 * @since V1.0
 */
@Component
public class SolrGatherScheduler {
    private Logger log = Logger.getLogger(SolrGatherScheduler.class);

    @Autowired
    private ZkNodeDAO zkNodeDAO;
    @Autowired
    private ZkNodeService zkNodeService;
    @Autowired
    private SolrNodeDAO solrNodeDAO;
    @Autowired
    private SolrNodeService solrNodeService;
    @Autowired
    private SolrCollectionDAO solrCollectionDAO;
    @Autowired
    private SolrCollectionService solrCollectionService;
    @Autowired
    private SolrShardDAO solrShardDAO;
    @Autowired
    private SolrShardService solrShardService;
    @Autowired
    private SolrReplicaDAO solrReplicaDAO;
    @Autowired
    private SolrReplicaService solrReplicaService;
    @Autowired
    private SolrCacheService solrCacheService;
    @Autowired
    private SolrNodeMonitorService solrNodeMonitorService;
    @Autowired
    private SolrReplicaMonitorService solrReplicaMonitorService;
    @Autowired
    private SolrCacheMonitorService solrCacheMonitorService;
    @Autowired
    private OperationLogService operationLogService;

    /**
     * 采集solr数据
     *
     * @throws Exception
     */
    public void execute() throws Exception {
        long nowTime = new Date().getTime();
        Integer backupId = 0;

        //更新SolrNode
        List<ZkNode> solrNodes = zkNodeDAO.findChildrenByPath("/overseer_elect/election", backupId);
        if (CollectionUtil.isEmpty(solrNodes)) {
            return;
        }
        Map<String, SolrNode> nodes = new HashMap<String, SolrNode>();
        List<Integer> disabledNodeIds = solrNodeDAO.findColumn(new Query(), SolrNode.ID, true, true);
        for (ZkNode solrNode : solrNodes) {
            SolrNode node = solrNodeService.saveByZKInfo(solrNode.getPath(), nowTime);
            nodes.put(node.getName(), node);
            disabledNodeIds.remove(node.getId());
        }
        if (disabledNodeIds.size() > 0) {
            solrNodeService.disable(disabledNodeIds);
        }

        //更新SolrCollection
        List<ZkNode> collectionNodes = zkNodeDAO.findChildrenByPath("/collections", backupId);
        if (CollectionUtil.isEmpty(collectionNodes)) {
            return;
        }
        Map<String, SolrCollection> collections = new HashMap<String, SolrCollection>();
        List<Integer> disableCollectionIds = solrCollectionDAO.findColumn(new Query(), SolrCollection.ID, true, true);
        Map<String, String> collectionUniqueKeyMap = zkNodeDAO.getCollectionUniqueKeyMap(backupId);
        for (ZkNode collectionNode : collectionNodes) {
            String collectionName = collectionNode.getRelativePath();
            SolrCollection collection = solrCollectionService.saveByZKInfo(collectionName, collectionNode.getPath(), collectionNode.getData(),
                    collectionUniqueKeyMap.get(collectionName), false, nowTime);
            collections.put(collection.getName(), collection);
            disableCollectionIds.remove(collection.getId());
        }
        if (disableCollectionIds.size() > 0) {
            solrCollectionService.disable(disableCollectionIds);
        }

        //更新SolrShard和SolrReplica
        ZkNode clusterStateNode = zkNodeDAO.getByPath("/clusterstate.json", backupId);
        JsonUtil.MapWrapper clusterStateMap = new JsonUtil.MapWrapper(JsonUtil.fromJsonAsMap(clusterStateNode.getData()));
        List<Integer> disableShardIds = solrShardDAO.findColumn(new Query(), SolrShard.ID, true, true);
        List<Integer> disableReplicaIds = solrReplicaDAO.findColumn(new Query(), SolrReplica.ID, true, true);
        for (String collectionName : clusterStateMap.keySet()) {
            SolrCollection collection = collections.get(collectionName);
            JsonUtil.MapWrapper collectionStateMap = new JsonUtil.MapWrapper(clusterStateMap.get(collectionName));

            //更新SolrCollection的router
            solrCollectionService.updateRouter(collection.getId(), collectionStateMap);

            //更新SolrShard
            boolean collectionAvailable = true;
            JsonUtil.MapWrapper shardStateMap = new JsonUtil.MapWrapper(collectionStateMap.get("shards"));
            for (String shardName : shardStateMap.keySet()) {
                SolrShard shard = solrShardService.saveByZKInfo(shardName, shardStateMap, collection.getId(), nowTime);
                collection.addShard(shard);
                disableShardIds.remove(shard.getId());

                if (!SolrState.isAlive(shard.getState())) {
                    collectionAvailable = false;
                }

                // 更新SolrReplica
                JsonUtil.MapWrapper replicaStateMap = new JsonUtil.MapWrapper(shardStateMap.getByPath(shardName, "replicas"));
                for (String replicaName : replicaStateMap.keySet()) {
                    SolrNode node = nodes.get(replicaStateMap.getStringByPath(replicaName, "node_name"));
                    if (node == null) {
                        continue;
                    }
                    SolrReplica replica = solrReplicaService.saveByZKInfo(replicaName, replicaStateMap.getMapWrapperByPath(replicaName), shard.getId(), collection.getId(), node.getId(), nowTime);
                    shard.addReplica(replica);
                    node.addReplica(replica);
                    disableReplicaIds.remove(replica.getId());
                }
            }

            solrCollectionService.updateAvailable(collection.getId(), collectionAvailable);
        }
        if (disableShardIds.size() > 0) {
            solrShardService.disable(disableShardIds);
        }
        if (disableReplicaIds.size() > 0) {
            solrReplicaService.disable(disableReplicaIds);
        }

        Map<Integer, Integer> nodeIds = new HashMap<Integer, Integer>();
        Map<Integer, Integer> collectionIds = new HashMap<Integer, Integer>();
        Map<Integer, Integer> replicaIds = new HashMap<Integer, Integer>();

        for (SolrNode node : nodes.values()) {
            nodeIds.put(node.getId(), node.getId());

            try {
                URLConnection connection = new URL(node.getMonitorUrl()).openConnection();
                String content = IOUtil.in2Str(connection.getInputStream(), "utf-8");
                JsonUtil.MapWrapper nodeMap = new JsonUtil.MapWrapper(JsonUtil.fromJsonAsMap(content));
                JsonUtil.MapWrapper statusMap = new JsonUtil.MapWrapper(nodeMap.get("status"));
                JsonUtil.MapWrapper cacheMap = new JsonUtil.MapWrapper(nodeMap.get("cache"));
                JsonUtil.MapWrapper luceneMap = new JsonUtil.MapWrapper(nodeMap.get("lucene"));
                JsonUtil.MapWrapper jvmMap = new JsonUtil.MapWrapper(nodeMap.get("jvm"));
                JsonUtil.MapWrapper systemMap = new JsonUtil.MapWrapper(nodeMap.get("system"));

                solrNodeService.updateBySolrInfo(node.getId(), luceneMap, jvmMap, systemMap);
                solrNodeMonitorService.createBySolrInfo(node.getId(), jvmMap, systemMap, nowTime);
                for (String coreName : statusMap.keySet()) {
                    SolrReplica replica = node.getReplicaByCoreName(coreName);
                    if (replica == null) {
                        continue;
                    }
                    replicaIds.put(replica.getId(), replica.getId());
                    collectionIds.put(replica.getCollectionId(), replica.getCollectionId());

                    solrReplicaService.updateBySolrInfo(replica.getId(), statusMap.getMapWrapperByPath(coreName));
                    solrReplicaMonitorService.createBySolrInfo(replica, replica.getState(), statusMap.getMapWrapperByPath(coreName), nowTime);
                    JsonUtil.MapWrapper coreCacheMap = cacheMap.getMapWrapperByPath(coreName);
                    for (String cacheType : coreCacheMap.keySet()) {
                        SolrCache cache = solrCacheService.saveBySolrInfo(replica.getId(), cacheType, coreCacheMap.getMapWrapperByPath(cacheType), nowTime);
                        solrCacheMonitorService.createBySolrInfo(replica, cacheType, coreCacheMap.getMapWrapperByPath(cacheType), nowTime);
                        replica.addCache(cache);
                    }
                }
            } catch (Exception e) {
                log.error(String.format("gather url[%s] failure", node.getMonitorUrl()));
                e.printStackTrace();
            }
        }

        operationLogService.createGatherLog(OperationType.GATHER_SOLR);
    }
}
